"""
Case Type   : 基础功能-SQL语法-DDL-ALTER TABLE PARTITION
Case Name   : 分区表锁降级，在两个session分别执行sql
Create At   : 2024/10/28
Owner       : liu-tong-8848
Description :
    1.创建分区表并在表内插入数据
    2.同时开两个会话，会话1执行drop partition，会话2执行select
    3.清理环境
Expect      :
    1.成功
    2.两个会话均执行成功
    3.清理环境成功
History     :
"""

import os
import unittest
import time

import py_opengauss
from yat.test import Node

from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread



class RangePartitionCase(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.user_node = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.table_name = 't_range_table_0084'
        self.user_param = f'-U {self.user_node.db_user} ' \
                          f'-W {self.user_node.db_password}'

    def test_partition_ddl_dml(self):
        step1 = '----step1:创建分区表并在表内插入数据;expect:成功----'
        self.log.info(step1)
        sql_cmd = "drop table if exists test_ugi_001; " \
                  + "create table test_ugi_001 ( c_id integer not null, c_name varchar(16) default 'omm', " \
                  + "c_class varchar(20) not null )partition by range(c_id) ( partition p1 values less than (1001), " \
                  + "partition p2 values less than (2001), partition p3 values less than (3001), " \
                  + "partition p4 values less than (4001), partition p5 values less than (5001), " \
                  + "partition p6 values less than (6001) );"
        result = self.primary_sh.execut_db_sql(sql_cmd, sql_type=self.user_param)
        self.log.info(result)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result, "执行失败:" + step1)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, result, "执行失败:" + step1)
        sql_cmd = "insert into test_ugi_001(c_id, c_class) select r, '1-1' from generate_series(1,1000) as r;" \
                  + "insert into test_ugi_001(c_id, c_class) select r, '1-2' from generate_series(1001,2000) as r;" \
                  + "insert into test_ugi_001(c_id, c_class) select r, '1-3' from generate_series(2001,3000) as r;" \
                  + "insert into test_ugi_001(c_id, c_class) select r, '1-4' from generate_series(3001,4000) as r;" \
                  + "insert into test_ugi_001(c_id, c_class) select r, '1-5' from generate_series(4001,5000) as r;" \
                  + "insert into test_ugi_001(c_id, c_class) select r, '1-6' from generate_series(5001,6000) as r; "
        result = self.primary_sh.execut_db_sql(sql_cmd, sql_type=self.user_param)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, "执行失败:" + step1)
        self.assertNotIn("ERROR", result, "执行失败:" + step1)

        step2 = '----step2:同时开两个会话，会话1执行drop partition，会话2执行select;expect:两个会话均执行成功----'
        self.log.info(step2)
        threads = []
        drop_partition_sql = f"begin; alter table test_ugi_001 drop partition p3;"
        thread1 = ComThread(self.primary_sh.execut_db_sql, args=(drop_partition_sql,))
        thread1.setDaemon(True)
        threads.append(thread1)
        select_sql = f"select * from test_ugi_001 where c_id = 15;"
        thread2 = ComThread(self.primary_sh.execut_db_sql, args=(select_sql,))
        thread2.setDaemon(True)
        threads.append(thread2)
        for thread in threads:
            thread.start()
        time.sleep(3)
        result1 = thread1.get_result()
        self.log.info(result1)
        self.assertIn(self.constant.ALTER_TABLE_MSG, result1, "执行失败:" + step2)
        self.assertNotIn("ERROR", result1, "执行失败:" + step2)
        result2 = thread2.get_result()
        self.log.info(result2)
        self.assertIn("15 | omm    | 1-1", result2, "执行失败:" + step2)
        self.assertIn("(1 row)", result2, "执行失败:" + step2)
        self.assertNotIn("ERROR", result2, "执行失败:" + step2)

    def tearDown(self):
        step3 = '--step3:清理环境;expect:清理环境成功--'
        self.log.info(step3)
        result = self.primary_sh.execut_db_sql(f'drop table test_ugi_001; ')
        self.log.info(result)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result, "执行失败:" + step3)
        self.log.info(f'-----{os.path.basename(__file__)} end--')
